package jmars.mobile.worker;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;

import jmars.mobile.common.Investigator;
import jmars.mobile.common.PropertiesLoader;
import jmars.mobile.common.Sha;
import jmars.mobile.datainvestigation.Summary;

/**
 * we should set
 * groupIndex = 1,2
 * batch=-1
 * @author Jan
 *
 */
public class CellularTopologies20 implements Investigator {
	private Set<String> nodes = new HashSet<String>();
	private Map<String,Map<Integer,Integer>> degrees = new HashMap<String,Map<Integer,Integer>>();
	private Map<Integer,Map<String,Set<String>>> edges = new HashMap<Integer,Map<String,Set<String>>>();

	private Set<String> lookupSet = new HashSet<String>();
	
	private Long computeAll(File[] files) throws IOException{
		Long interval = 20*60000L;
		Integer batch = Integer.parseInt(PropertiesLoader.getProperty("batch"));
		int l=(files.length>batch&&batch>0)?batch:files.length; //The max file count will be processed. This is a performance threshold.
		Long buffer = 10*60*1000L;
		Long total = 0L;
		Long line;
		int filesFinished=0;
		FileInputStream fis;
		BufferedReader reader;
		String row;
		String[] cols;
		Map<String,Map<Long,String>> phoneRecords=new HashMap<String,Map<Long,String>>();
		edges.put(20, new HashMap<String,Set<String>>());
		Resource r=new ClassPathResource("lookup-yuzhong.txt");
		BufferedReader readerLookup = new BufferedReader(new InputStreamReader(r.getInputStream()));
		while((row=readerLookup.readLine())!=null){
			if(row!=null)lookupSet.add(row);
		}
		System.out.println("lookup set: "+lookupSet.size());
		Long beginL=null,beginR=null;
		Calendar cal=Calendar.getInstance();
		Date beginDate=null;
		DecimalFormat df = new DecimalFormat();
		df.applyPattern("0.000");
		String groupKey=null;
		ArrayList<Entry<Long,String>> tempList;
		Entry<Long,String> lastEntry=null;
		File inputFile;
		for(int i=0;i<l;i++){
			inputFile=files[i];
			fis = new FileInputStream(inputFile);
			reader = new BufferedReader(new InputStreamReader(fis));
			line=0L;
			while((row=reader.readLine())!=null){
				// construct the phone records.
				line++;
				total++;
				try{
					cols=row.split(Summary.columnSeperator);
					if(Summary.filterRowExcludeLACI(cols,true,false)) continue;
					beginDate=Summary.parseBeginDate(cols);
					if(beginDate==null) continue;
					cal.setTime(beginDate);
					if(cal.get(Calendar.YEAR)!=2012) continue;
					if(beginL==null)beginL=beginDate.getTime();
					if(beginR==null)beginR=beginDate.getTime();
					if(Summary.checkLACI(cols))
						groupKey=Summary.genGroupKey(cols);
					else
						groupKey=null;
					if(!phoneRecords.containsKey(cols[0]))
						phoneRecords.put(cols[0], new HashMap<Long,String>());
					phoneRecords.get(cols[0]).put(beginDate.getTime(), groupKey);
					if(groupKey!=null&&lookupSet.contains(Sha.digest32HexSHA1(groupKey))){
						if(!nodes.contains(groupKey))nodes.add(groupKey);
						if(!degrees.containsKey(groupKey)){
							degrees.put(groupKey, new HashMap<Integer,Integer>());
							if(!edges.get(20).containsKey(groupKey))edges.get(20).put(groupKey, new HashSet<String>());
						}
					}
				}catch(Exception ex){
					System.out.println("exception at line "+line+" of "+inputFile.getName());
					System.out.println(row);
					ex.printStackTrace();
					if(!Summary.errorResume){
						System.exit(0);
					}
				}
			}
			if((beginR!=null&&(beginR-beginL>interval+buffer))||i==l-1){
				//work
				for(String phone:phoneRecords.keySet()){
					tempList = new ArrayList<Entry<Long,String>>(phoneRecords.get(phone).entrySet());
					Collections.sort(tempList, new Comparator<Map.Entry<Long,String>>() {
						public int compare(Map.Entry<Long, String> o1, Map.Entry<Long, String> o2) {
							return o2.getKey()==o1.getKey()?0:(o2.getKey()<o1.getKey()?1:-1);
			            }
					});
					lastEntry=null;
					for(Entry<Long,String> en:tempList){
						if(i<l-1&&beginR-en.getKey()<buffer) break;
						
						if(lastEntry!=null){
							if(lastEntry.getValue()!=null&&en.getValue()!=null&&!lastEntry.getValue().equals(en.getValue())&&
									(lookupSet.contains(Sha.digest32HexSHA1(lastEntry.getValue()))||lookupSet.contains(Sha.digest32HexSHA1(en.getValue())))){
								if(en.getKey()-lastEntry.getKey()<1000*60*20){
									if(edges.get(20).containsKey(lastEntry.getValue()))edges.get(20).get(lastEntry.getValue()).add(en.getValue());
									if(edges.get(20).containsKey(en.getValue()))edges.get(20).get(en.getValue()).add(lastEntry.getValue());
								}
							}
						}
						lastEntry=en;

					}
				}
				beginR=null;
				beginL=null;
				phoneRecords.clear();
			}
			beginR=null;
			filesFinished++;
			reader.close();
			fis.close();
			System.out.println(filesFinished+": "+inputFile.getName()+" finished at "+line);
		}

		for(String n:nodes){
			degrees.get(n).put(20, edges.get(20).get(n).size());
		}

		return total;
	}
	
	@Override
	public Long compute(File[] files, FileWriter writer) throws IOException {
		Long total = 0L;
		total=computeAll(files);
		writer.write("%total: "+total+"\r\n%time");
		writer.write("%*****************nodes and degrees******************\r\n");
		writer.write("%nodes,20mins\r\n");
		for(String n:nodes){
			writer.write(Sha.digest32HexSHA1(n)+","+degrees.get(n).get(20)+"\r\n");
		}
		writer.write("%*********************20 mins***************************\r\n");
		for(String ek:edges.get(20).keySet()){
			if(edges.get(20).get(ek).isEmpty())continue;
			for(String te:edges.get(20).get(ek)){
				if(lookupSet.contains(Sha.digest32HexSHA1(te))){
					writer.write(Sha.digest32HexSHA1(ek));
					writer.write(","+Sha.digest32HexSHA1(te)+"\r\n");
				}
			}
		}
		return total;
	}

}
